package com.lib76.test.common.zookeeper;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import com.gzhryc.common.logging.Logger;

public class ZookeeperConfigHelper implements Watcher {

	final static Logger log = Logger.getLogger(ZookeeperConfigHelper.class);

	static ZookeeperConfigHelper me;

	public static ZookeeperConfigHelper self() {
		if (me == null) {
			me = new ZookeeperConfigHelper();
		}
		return me;
	}

	String instanceId;
	ZooKeeper zookeeper;
	Pattern pat;

	Map<Long, AtomicBoolean> leaders;

	public ZookeeperConfigHelper() {
		this.leaders = new ConcurrentHashMap<Long, AtomicBoolean>();
		this.pat = Pattern.compile("^/iotNode/(\\d*)_leader$");
	}

	public ZookeeperConfigHelper initial(String ip, Integer port, Integer timeout, String instanceId) {
		try {
			this.instanceId = instanceId;
			this.zookeeper = new ZooKeeper(ip + ":" + port, timeout, this);
		} catch (IOException e) {
			log.info("连接Zookeeper错误", e);
		}
		return this;
	}

	public void registerLeader(Long iotId) {
		AtomicBoolean isLeader = this.leaders.get(iotId);
		if (!this.leaders.containsKey(iotId)) {
			isLeader = new AtomicBoolean(false);
			this.leaders.put(iotId, isLeader);
		}
		String path = "/iotNode";
		try {
			Stat stat = zookeeper.exists(path, false);
			if (stat == null) {
				zookeeper.create(path, iotId.toString().getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			} else {
				zookeeper.setData(path, iotId.toString().getBytes(), -1);
			}
		} catch (InterruptedException e) {
			log.error("向Zookeeper注册Leader错误", e);
		} catch (KeeperException e) {
			log.error("向Zookeeper注册Leader错误", e);
		}

		path = path + "/" + iotId + "_leader";
		try {
			Stat stat = zookeeper.exists(path, false);
			if (stat == null) {
				String result = zookeeper.create(path, instanceId.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
				log.info("向Zookeeper注册Leader成功，结果：" + result);
				isLeader.set(true);
			}
		} catch (InterruptedException e) {
			log.error("向Zookeeper注册Leader错误", e);
		} catch (KeeperException e) {
			log.error("向Zookeeper注册Leader错误", e);
		}

		if (!isLeader.get()) {
			Stat stat = new Stat();
			try {
				byte[] data = zookeeper.getData(path, true, stat);
				log.info("未获取Leader，节点" + iotId + "的Leader实例为：" + new String(data));
			} catch (KeeperException e) {
				log.error("向Zookeeper获取" + path + "数据错误", e);
			} catch (InterruptedException e) {
				log.error("向Zookeeper获取" + path + "数据错误", e);
			}
		}
	}

	public boolean isLeader(Long iotId) {
		AtomicBoolean isLeader = this.leaders.get(iotId);
		if (isLeader != null) {
			return isLeader.get();
		} else {
			log.error(iotId + "节点未找到Leader检查对象");
		}
		return false;
	}

	@Override
	public void process(WatchedEvent event) {
		switch (event.getType()) {
		case NodeCreated:
			log.info("Zookeeper节点创建事件：" + event.getPath());
			break;
		case NodeDeleted:
			log.info("Zookeeper节点删除事件：" + event.getPath());
			Matcher mat = pat.matcher(event.getPath());
			if (mat.find()) {
				String temp = mat.group(1);
				// 触发重新选举
				registerLeader(Long.valueOf(temp));
			}
			break;
		case NodeDataChanged:
			log.info("Zookeeper节点数据更新事件：" + event.getPath());
			break;
		default:
			log.info("Zookeeper事件：eventType = " + event.getType().name());
			break;
		}
	}

	public void delLeader(Long iotId) {
		String path = "/iotNode/" + iotId + "_leader";
		try {
			zookeeper.delete(path, -1);
		} catch (InterruptedException e) {
			log.error("Zookeeper删除" + iotId + "节点Leader错误", e);
		} catch (KeeperException e) {
			log.error("Zookeeper删除" + iotId + "节点Leader错误", e);
		}
	}

	public void close() {
		try {
			zookeeper.close();
			log.info(instanceId + "实例关闭Zookeeper连接");
		} catch (InterruptedException e) {
			log.error("Zookeeper关闭错误", e);
		}
	}

}
